package com.shiku.push;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import cn.xyz.commons.support.mongo.MongoOperator;
import cn.xyz.commons.utils.StringUtil;
import com.mongodb.BasicDBObject;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.ServiceState;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.jivesoftware.smack.SmackException;
import org.jivesoftware.smack.XMPPException;
import org.jivesoftware.smack.chat2.Chat;
import org.jivesoftware.smack.packet.Message;
import org.jxmpp.jid.impl.JidCreate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.tio.core.ChannelContext;

import com.alibaba.fastjson.JSON;
import com.shiku.imclient.BaseClientHandler;
import com.shiku.imclient.BaseClientListener;
import com.shiku.imclient.BaseIMClient;
import com.shiku.imserver.common.message.AuthMessage;
import com.shiku.imserver.common.message.ChatMessage;
import com.shiku.imserver.common.message.MessageHead;
import com.shiku.imserver.common.packets.ChatType;
import com.shiku.imserver.common.utils.StringUtils;
import com.shiku.xmpppush.autoconfigure.ApplicationProperties.IMConfig;
import com.shiku.xmpppush.autoconfigure.ApplicationProperties.MQConfig;

import cn.xyz.commons.utils.DateUtil;
import cn.xyz.mianshi.service.impl.RoomManagerImplForIM;
import cn.xyz.mianshi.service.impl.UserManagerImpl;
import cn.xyz.mianshi.utils.SKBeanUtils;
import cn.xyz.service.KXMPPServiceImpl.MessageBean;
import lombok.Getter;
import lombok.Setter;

import javax.annotation.Resource;

@Component
@RocketMQMessageListener(topic = "xmppMessage", consumerGroup = "xmppNewProducer")
public class SkMessageListenerConcurrently  implements RocketMQListener<MessageExt>,InitializingBean {
	
	private static final Logger log = LoggerFactory.getLogger(SkMessageListenerConcurrently.class);
	
	private static RoomManagerImplForIM getRoomManager(){
		RoomManagerImplForIM roomManager = SKBeanUtils.getRoomManagerImplForIM();
		return roomManager;
	};
	private static UserManagerImpl getUserManager(){
		UserManagerImpl userManager = SKBeanUtils.getUserManager();
		return userManager;
	};
	
	private String name_addr="";
	
	private Map<String,ChatMessageVo> messageMap=new ConcurrentHashMap<String,ChatMessageVo>();
	
	/**
	 * 获取系统号
	 */
	private List<String> sysUserList=null;

	@Autowired(required=false)
	public IMConfig imConfig;
	
	private synchronized List<String> getUserList(){
		if(null!=sysUserList)
			return sysUserList;
		sysUserList=Collections.synchronizedList(new ArrayList<String>());
		/*for (String string : SKBeanUtils.getSystemAdminMap().keySet()) {
			sysUserList.add(string);
		}*/
		return sysUserList;
	}
	
	//public MQConfig mqConfig;
	
	private Map<String,IMClient> clientMap=new ConcurrentHashMap<>();


	@Resource
	private RocketMQTemplate rocketMQTemplate;
	
	// 新的队列
	private ConcurrentLinkedQueue<ChatMessage> queue = new ConcurrentLinkedQueue<ChatMessage>();
	@Override
	public void afterPropertiesSet() throws Exception {
		initThread();

	}
	public  void initThread(){
		ImPushQueueThread work =null;
		IMClient client=null;
		/*for(int i=0;i<getUserList().size();i++){
			if(getUserList().get(i)=="10000"){
				continue ;
			}*/
		try {
			client =getIMClient("9999");
		} catch (Exception e) {
			e.printStackTrace();
		}
		work=new ImPushQueueThread(client);
		work.start();
		/*}*/
		timer();
	}


	@Override
	public void onMessage(MessageExt messageExt) {
		MessageBean messageBean = null;
		ChatMessage message=null;

		String body=null;
		try {
			body=new String(messageExt.getBody(),"utf-8");
			log.info("new msg {}",body);
			messageBean=JSON.parseObject(body, MessageBean.class);
			if(2==messageBean.getMsgType()){
				sendBroadCast(messageBean);
				return;
			}
			message=bulidChatMessage(messageBean);
			if(null==message)
				return;
			queue.offer(message);
		} catch (Exception e) {
			log.error("=== "+body+" ===> "+e.getMessage());
		}
	}
	public ChatMessage bulidChatMessage(MessageBean messageBean){
		ChatMessage message=null;
		MessageHead messageHead=null;
		try {
			message=new ChatMessage();
			messageHead=new MessageHead();
			if(!StringUtil.isEmpty(messageBean.getTo())){
				messageHead.setTo(messageBean.getTo());
			}else{
				messageHead.setTo(messageBean.getToUserId());
			}

			byte chatType=ChatType.CHAT;
			if(1==messageBean.getMsgType()) {
				chatType=ChatType.GROUPCHAT;
				messageHead.setTo(messageBean.getRoomJid());
				//messageBean.setToUserId(messageBean.getRoomJid());
			}else if(2==messageBean.getMsgType()) {
				chatType=ChatType.ALL;
			}

			messageHead.setChatType(chatType);
			if(null!=messageBean.getMessageId())
				messageHead.setMessageId(messageBean.getMessageId());
			else {
				messageHead.setMessageId(StringUtils.newStanzaId());
			}
			if(null!=messageBean.getContent())
				message.setContent(messageBean.getContent().toString());
			message.setFromUserId(messageBean.getFromUserId());
			message.setFromUserName(messageBean.getFromUserName());
			message.setToUserId(messageBean.getToUserId());
			message.setToUserName(messageBean.getToUserName());
			message.setType((short)messageBean.getType());
			long timeSend = System.currentTimeMillis();
			messageBean.setTimeSend(timeSend);
			message.setTimeSend(timeSend);
			if(null!=messageBean.getObjectId())
				message.setObjectId(messageBean.getObjectId().toString());
			if(null!=messageBean.getFileName())
				message.setFileName(messageBean.getFileName());
			message.setMessageHead(messageHead);
		} catch (Exception e) {
			log.error("bulidChatMessage {}",e.getMessage());
		}
		return message;
	}
	public void sendBroadCast(MessageBean messageBean){
		List<Integer> list=new ArrayList<>();
		list = SKBeanUtils.getRedisService().getFriendsUserIdsList(Integer.valueOf(messageBean.getFromUserId()));
		if(list.size()==0){
			list = SKBeanUtils.getFriendsManager().queryFansId(Integer.valueOf(messageBean.getFromUserId()));
			SKBeanUtils.getRedisService().saveFriendsUserIdsList(Integer.valueOf(messageBean.getFromUserId()),list);
		}
		for(Integer userId:list){
			ChatMessage message=null;
			MessageHead messageHead=null;
			try {
				message=new ChatMessage();
				messageHead=new MessageHead();
				messageHead.setFrom(messageBean.getFromUserId());
				messageHead.setTo(userId.toString());
				messageHead.setChatType(ChatType.CHAT);
				message.setFromUserId(messageBean.getFromUserId());
				message.setFromUserName(messageBean.getFromUserName());
				message.setToUserId(userId.toString());
				message.setToUserName(messageBean.getToUserName());
				message.setType((short)messageBean.getType());

				message.setTimeSend(System.currentTimeMillis());
				message.setContent(messageBean.getContent().toString());
				if(null!=messageBean.getMessageId())
					messageHead.setMessageId(messageBean.getMessageId());
				else {
					messageHead.setMessageId(StringUtils.newStanzaId());
				}
				message.setMessageHead(messageHead);
				// 把消息丢进queue队列中
				queue.offer(message);
			} catch (Exception e) {
				e.printStackTrace();
				log.error("放进队列失败!");
			}

		}
	}
	
	
	private IMClient getIMClient(String userId) {
		IMClient client=new IMClient();
		client.setUserId(userId);
		
		client.setPingTime(imConfig.getPingTime());
		
		BaseClientHandler clientHandler=new BaseClientHandler() {
			
			@Override
			public void handlerReceipt(String messageId) {
				//System.out.println("handlerReceipt ===> "+messageId);
				messageMap.remove(messageId);
			}
		};
		BaseClientListener clientListener=new BaseClientListener() {
			
			@Override
			public AuthMessage authUserMessage(ChannelContext channelContext,BaseIMClient client) {
				MessageHead messageHead=new MessageHead();
				
				messageHead.setChatType((byte)1);
				channelContext.userid=userId;
				messageHead.setFrom(userId+"/Server");
				
				AuthMessage authMessage=new AuthMessage();
				authMessage.setPassword("123456");
				authMessage.setToken("token");
				authMessage.setMessageHead(messageHead);
				return authMessage;
			}
		}; 
		
		client.initIMClient(imConfig.getHost(),imConfig.getPort(),clientHandler,clientListener);
		
		return client;
	}
	
	
	/**
	 * 将消息重新放入队列
	 * @param messageBean
	 */
	public synchronized void sendAgainToMQ(MessageBean messageBean){
		try {
			rocketMQTemplate.syncSend("xmppMessage", messageBean.toString());
		} catch (Exception e) {
			log.error("重新放入队列失败");
			e.printStackTrace();
		}
	}
	

	/**
	 * 推送Queue队列中的消息
	 * @param conn
	 * @throws InterruptedException 
	 * @throws XMPPException 
	 * @throws IOException 
	 * @throws SmackException 
	 */
	public void runQueuePush(IMClient client)
			throws Exception{
		ChatMessage message=queue.poll();
		
		if(message==null){
			return;
		}
		try {
			if(null==client)
				Thread.sleep(500);
			message.getMessageHead().setFrom(client.getUserId()+"/Server");
			ChatMessageVo messageVo=new ChatMessageVo();
			messageVo.setCreateTime(DateUtil.currentTimeSeconds());
			messageVo.setMessage(message);		
			client.sendMessage(message);
			messageMap.put(message.getMessageHead().getMessageId(), messageVo);
			log.info("系统推送成功： to {},",message.getToUserId());
		}  catch (Exception e) {
			queue.offer(message);
			log.error(" 发送推送失败!=====> {}",e.getMessage());
		}
	}



	/**
	 * 
	 * @Description: TODO(在线程中消费队列中的消息)
	 * @author Administrator
	 * @date 2018年12月26日 上午11:26:22
	 * @version V1.0
	 */
	public class ImPushQueueThread extends Thread {
		private IMClient client=null;
		
		public ImPushQueueThread() {}
		
		public ImPushQueueThread(IMClient client) {
			this.client=client;
		}
		
        @Override
        public void run() {
        	while (true) {
        		if(!queue.isEmpty()){
        			try {
						runQueuePush(client);
					} catch (Exception e) {
						e.printStackTrace();
					}
        		}else{
        			try {
        				Thread.sleep(100);
					} catch (Exception e) {
						e.printStackTrace();
					}
        		}
            }
        }

    }

	
	@Getter
	@Setter
	public class ChatMessageVo {
		private long createTime;
		private ChatMessage message;
	}
	
	
	/**
	 * 定时重发
	 * @throws InterruptedException 
	 */
	public void timer(){
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				while(true){
					Long startTime=System.currentTimeMillis();
					if(messageMap.size()>10){
						log.info("开始时间"+DateUtil.currentTimeSeconds()+"   map大小   "+messageMap.size());
					}
					Set<Entry<String, ChatMessageVo>> set=messageMap.entrySet();
					for (Entry<String, ChatMessageVo> entry : set) {
						ChatMessageVo messageVo=entry.getValue();
						if(messageVo!=null){
							if(DateUtil.currentTimeSeconds()-messageVo.getCreateTime()>=30){
								queue.offer(messageVo.getMessage());
							}
						}else {
							return;
						}
					}
					
					Long endTime=System.currentTimeMillis();
					if((endTime-startTime)>1000){
						log.info("执行map所需要的时间========"+(endTime-startTime));
					}
					try {
						Thread.sleep(30000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				
			}

		}).start();
		
	}
}
